package tul.cot.lib.reducer;

import java.util.List;

import tul.cot.io.Header;
import tul.cot.io.PlotPoint;
import tul.cot.job.Migrations.TaskReducer;
import tul.cot.util.Histogram;
import tul.cot.util.HistogramAggregator;
import tul.cot.util.MigAggregator;


public class MigReducer extends TaskReducer {
	
	@Override
	protected void reduce(Header key, Iterable<PlotPoint> values, Context context)
		throws java.io.IOException ,InterruptedException {
		
		if (key.getSegment() < 0) {
			
			HistogramAggregator aggregator = new HistogramAggregator(NUM_BUCKETS);
			aggregator.aggregate(values);
			
			List<Histogram> histograms = aggregator.getHistograms();
			for (Histogram h: histograms) {
				emit_Histogram(context, h);
			}
			
		} else {
			
			MigAggregator aggregator = new MigAggregator(NUM_TREES, HPD_CONF_LEVEL);
			aggregator.aggregate(values);
			
			List<Double> medianValues = aggregator.getMedianValues();
			List<Double> hpdMinValues = aggregator.getHpdMinValues();
			List<Double> hpdMaxValues = aggregator.getHpdMaxValues();

			emit_Values(context, key, MEDIAN, medianValues);
			emit_Values(context, key, HPD_MIN, hpdMinValues);
			emit_Values(context, key, HPD_MAX, hpdMaxValues);
		}
	}
}
